package com.shujia.spark.stream

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Duration, Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object Demo1WordCount {
  def main(args: Array[String]): Unit = {


    /**
      * 1、创建spark环境
      *
      */
    val conf = new SparkConf()

    conf.setMaster("local[2]")
    conf.setAppName("wc")

    val sc = new SparkContext(conf)


    /**
      * 创建SparkStreaming环境
      *  Durations.seconds(5): 多久执行一次
      *
      */

    val ssc = new StreamingContext(sc, Durations.seconds(5))


    /**
      * 1、读取数据
      * 读取socket中额数据
      *
      */


    val linesDS: ReceiverInputDStream[String] = ssc.socketTextStream("master", 8888)

    val wordsDS: DStream[String] = linesDS.flatMap(_.split(","))

    val kvDS: DStream[(String, Int)] = wordsDS.map((_, 1))

    val countDS: DStream[(String, Int)] = kvDS.reduceByKey(_ + _)

    countDS.print()

    /**
      * 需要启动陈鼓型
      */

    //启动
    ssc.start()
    //等待关闭
    ssc.awaitTermination()
    //关闭
    ssc.stop()


  }

}
